package com.kafka.kafka;

import com.kafka.entity.Message;
import com.kafka.logs.SystemLogger;
import com.kafka.model.TcpServer;
import com.kafka.utils.ProtoStuffSerializeUtils;
import io.netty.channel.Channel;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @ClassName KafkaConsumer
 * @Description TODD
 * @Author Jesson
 * @Date 2019/1/15 9:05
 * @Version 1.0
 **/
@Component
public class KafkaConsumer {
    @KafkaListener(topics = "${kafka.consumer.topic}",containerFactory = "kafkaListenerContainerFactory")
    public void listen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        try {
            for (ConsumerRecord<?, ?> record : records) {
                byte[] msg = (byte[])record.value();
                if (null != msg) {
                    Message message = ProtoStuffSerializeUtils.deSerialize(msg,Message.class);
                    if(message == null){
                        continue;
                    }
                    Channel ch = TcpServer.getChannelMap().get(message.getId());
                    if(ch != null && ch.isActive()){
                        //寻找客户端连接，回复响应数据
                        ch.writeAndFlush(message.getMsg());
                        SystemLogger.info("network reply success,ip:{},uuid:{},messageLen:{}",
                                message.getAddr(),message.getId(),message.getMsg().length);
                    }else{
                        if(ch != null) {
                           SystemLogger.error("network reply failed,terminal network has been disconnected,ip:{}",ch.remoteAddress().toString());
                        }
                    }
                    TcpServer.getChannelMap().remove(message.getId());
                }
            }
        } catch (Exception e) {
           SystemLogger.error("KafkaConsumer exception:{}",e.getMessage());
        }finally{
            //提交偏移量
            ack.acknowledge();
        }
    }
}
